package com.avoid.easymqtt.broker;

import com.avoid.easymqtt.common.RejectHandler;
import com.avoid.easymqtt.common.ThreadFactoryImpl;
import com.avoid.easymqtt.mqtt.MqttRequestProcessor;
import com.avoid.easymqtt.mqtt.protocol.ConnectProcessor;
import com.avoid.easymqtt.remoting.RemotingServer;
import com.avoid.easymqtt.remoting.netty.NettyRemotingServer;
import com.avoid.easymqtt.remoting.netty.NettyServerConfig;
import io.netty.handler.codec.mqtt.MqttMessageType;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 描述
 *
 * @author avoid@fiture.com
 * @date 2022/5/9
 * @motto Life is so short,do something to make yourself happy,such as coding
 */
public class BrokerController {

    private final BrokerConfig brokerConfig;
    private final NettyServerConfig nettyServerConfig;
    private RemotingServer remotingServer;
    private final BlockingQueue<Runnable> connectThreadPoolQueue;
    private ExecutorService connectMessageExecutor;

    public BrokerController(final BrokerConfig brokerConfig, final NettyServerConfig nettyServerConfig) {
        this.brokerConfig = brokerConfig;
        this.nettyServerConfig = nettyServerConfig;
        this.connectThreadPoolQueue =
            new LinkedBlockingQueue<Runnable>(this.brokerConfig.getConnectThreadPoolQueueCapacity());
    }

    public boolean initialize() throws CloneNotSupportedException {
        this.remotingServer = new NettyRemotingServer(this.nettyServerConfig, null);

        this.connectMessageExecutor = new ThreadPoolExecutor(
            this.brokerConfig.getConnectMessageThreadPoolNums(),
            this.brokerConfig.getConnectMessageThreadPoolNums(),
            1000 * 60,
            TimeUnit.MILLISECONDS,
            connectThreadPoolQueue,
            new ThreadFactoryImpl("ConnectThread"), new RejectHandler("connect", connectThreadPoolQueue.size()));
        this.registerProcessor();
        return true;
    }

    public void registerProcessor() {
        MqttRequestProcessor connectProcessor = new ConnectProcessor();

        this.remotingServer.registerProcessor(MqttMessageType.CONNECT.value(), connectProcessor, this.connectMessageExecutor);
    }

    public String getBrokerAddr() {
        return this.brokerConfig.getBrokerIP() + ":" + this.nettyServerConfig.getListenPort();
    }

    public void start() throws Exception {
        if (this.remotingServer != null) {
            this.remotingServer.start();
        }
    }

    public void shutdown() {
        if (this.remotingServer != null) {
            this.remotingServer.shutdown();
        }
    }

    public BrokerConfig getBrokerConfig() {
        return brokerConfig;
    }
}
